package consume_sync

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/segmentio/kafka-go"
	"go-demo-2025/common"
	"go-demo-2025/utils/kafkaUtil"
	"log"
	"time"
)

// OrderMessage 订单消息结构体
type OrderMessage struct {
	OrderID string `json:"order_id"`
	Amount  int64  `json:"amount"`
	Status  string `json:"status"`
}

// 启动同步消费者
func StartSyncConsumer(ctx context.Context, topic string) error {
	// 创建Kafka读者（优化配置：指定分区+MinBytes=0，确保消息及时返回）
	reader := kafka.NewReader(kafka.ReaderConfig{
		Brokers:     kafkaUtil.KafkaBrokers,
		Topic:       topic,
		Partition:   0, // 固定消费第0分区（测试/单分区场景推荐）
		MinBytes:    0, // 有消息立即返回，避免阻塞
		MaxBytes:    10 * 1024 * 1024,
		StartOffset: kafka.FirstOffset,
		GroupID:     "order_sync_consumer_group",
	})
	defer reader.Close()

	// 增强启动日志
	log.Printf(" 同步消费者启动成功！")
	log.Printf(" 消费配置：主题=%s, 分区=0, 消费者组=%s", topic, "order_sync_consumer_group")
	log.Printf(" 等待接收新消息...\n")

	for {
		select {
		case <-ctx.Done():
			log.Println("\n 消费者收到退出信号，正在关闭...")
			return nil
		default:
			// 读取消息（添加超时日志，明确等待状态）
			log.Printf(" 正在拉取消息...")
			msg, err := reader.ReadMessage(ctx)
			if err != nil {
				log.Printf(" 读取消息失败: %v，1秒后重试...\n", err)
				time.Sleep(1 * time.Second)
				continue
			}

			// 增强消息接收日志
			log.Printf("\n 成功拉取到消息：")
			log.Printf("   偏移量: %d", msg.Offset)
			log.Printf("   消息Key: %s", string(msg.Key))
			log.Printf("   消息内容: %s", string(msg.Value))

			// 处理消息（带重试）
			if err := processWithRetry(ctx, msg); err != nil {
				log.Printf(" 消息最终处理失败（order_id=%s），发送到死信队列: err=%v\n", string(msg.Key), err)
				if err := sendToDeadLetter(ctx, msg, err.Error()); err != nil {
					log.Printf(" 发送死信队列失败: %v\n", err)
				}
			} else {
				// 处理成功后手动提交偏移量
				if err := reader.CommitMessages(ctx, msg); err != nil {
					log.Printf(" 偏移量提交失败（offset=%d）: %v\n", msg.Offset, err)
				} else {
					log.Printf(" 消息处理完成并提交！")
					log.Printf("   已提交偏移量: %d", msg.Offset)
					log.Printf("   处理订单ID: %s\n", string(msg.Key))
					log.Printf(" 等待接收新消息...\n")
				}
			}
		}
	}
}

// 带重试的消息处理
func processWithRetry(ctx context.Context, msg kafka.Message) error {
	var orderMsg OrderMessage
	if err := json.Unmarshal(msg.Value, &orderMsg); err != nil {
		return fmt.Errorf("解析消息失败: %v", err)
	}

	// 增强解析日志
	log.Printf(" 消息解析成功：order_id=%s, amount=%d, 原始状态=%s",
		orderMsg.OrderID, orderMsg.Amount, orderMsg.Status)

	// 最多重试3次
	retryCount := 0
	maxRetry := 3
	var lastErr error

	for retryCount < maxRetry {
		log.Printf(" 开始处理订单（第%d次尝试）: order_id=%s", retryCount+1, orderMsg.OrderID)
		if err := processOrderMessage(ctx, orderMsg); err != nil {
			lastErr = err
			retryCount++
			// 指数退避重试
			sleepTime := time.Duration(retryCount) * 1 * time.Second
			log.Printf(" 第%d次处理失败: err=%v, 等待%v后重试...",
				retryCount, err, sleepTime)
			time.Sleep(sleepTime)
			continue
		}
		log.Printf(" 订单处理成功（order_id=%s）", orderMsg.OrderID)
		return nil // 处理成功
	}

	return fmt.Errorf("超过最大重试次数(%d), 最后错误: %v", maxRetry, lastErr)
}

// 实际业务处理逻辑（新增订单状态更新）
func processOrderMessage(ctx context.Context, order OrderMessage) error {
	// 1. 模拟业务处理（如调用支付回调、库存扣减等）
	log.Printf(" 正在执行订单业务逻辑：order_id=%s", order.OrderID)
	time.Sleep(500 * time.Millisecond)

	// 2. 更新订单状态为"已支付"（核心闭环逻辑）
	updateSQL := `UPDATE orders 
	              SET status = 'paid', update_time = NOW() 
	              WHERE order_id = ? AND status = 'pending'` // 仅更新待支付状态的订单

	result, err := common.SqlDB.ExecContext(ctx, updateSQL, order.OrderID)
	if err != nil {
		return fmt.Errorf("更新订单状态失败: %v", err)
	}

	// 3. 验证更新结果
	rowsAffected, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("查询更新行数失败: %v", err)
	}
	if rowsAffected == 0 {
		return fmt.Errorf("未找到待更新的订单（order_id=%s）或订单已被处理", order.OrderID)
	}

	// 增强状态更新日志
	log.Printf(" 订单状态更新成功：order_id=%s, 旧状态=pending, 新状态=paid", order.OrderID)
	return nil
}

// 发送到死信队列
func sendToDeadLetter(ctx context.Context, msg kafka.Message, errMsg string) error {
	deadLetterTopic := "order_dead_letter"
	// 构造死信消息（包含原始消息和错误信息）
	deadLetterValue := fmt.Sprintf(`{
		"original_key": "%s",
		"original_value": "%s",
		"error": "%s",
		"timestamp": %d
	}`, string(msg.Key), string(msg.Value), errMsg, time.Now().UnixMilli())

	log.Printf(" 发送死信消息：topic=%s, key=%s", deadLetterTopic, string(msg.Key))
	return kafkaUtil.SendKafkaMessage(ctx, deadLetterTopic, string(msg.Key), deadLetterValue)
}
